//****************************************************************************
//File: Slave.java
//This Java Source File was implemented in the same way as Prof. Alan Kaminsky
//did in the Publisher.java and Subscriber.java class of his lecture notes.
//URL: http://www.cs.rit.edu/~ark/730/jms/jms.shtml
//The Slave program subscribe to the input topic of matrix multiplication
//and generates result column of the matrix multiplication and publishes to 
//the result topic.
//***************************************************************************
import java.io.Serializable;

import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.MessageProducer;
import javax.jms.ObjectMessage;
import javax.jms.Session;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;

/**
 * Class Slave subscribes to the input topic to get the
 * two matrices to be multiplied and calculates certain columns
 * of the resultant matrix based on its id and publishes it to
 * the result topic.
 *
 * Usage: java Slave <I>slaveid</I> <I>host</I> <I>port</I>
 * <BR><I>slaveid</I> = ID of the slave.
 * <BR><I>host</I> = Host name of the broker service
 * <BR><I>port</I> = Port number of the broker service
 *
 * @author  Vivek Narang (vsn6751@rit.edu)
 * @author  Alan Kaminsky
 * @version 05-May-2012
 */
public class Slave {
// Data Members	
	// Default user and password for the MQ Connection object
	private static String user = ActiveMQConnection.DEFAULT_USER;
	private static String password = ActiveMQConnection.DEFAULT_PASSWORD;
	// URL at which the connection is made	
	private static String url;
	// Session object
	private static Session session;
	// Message Producer
	private static MessageProducer producer;
	// Topic where input matrices are published
	private static String inputTopic = "MatrixMultiplication";
	// Topic where resultant column is published
	private static String resultTopic = "Result";
	// Slave ID
	private static int slaveId;
	// Message Consumer
	private static MessageConsumer consumer;
	// Connection Object
	private static Connection connection;
	
// Main Program
	/** 
	 * Main Program
	 */
	public static void main(String args[]) throws Exception {
		// Check for command line arguments
		if(args.length != 3) usage();
		
		slaveId = parseInt(args[0], "slaveid");
		String host = args[1];
		int port = parseInt(args[2], "");

		
		// Create connection
		url = "failover:tcp://" + host + ":" + port;
		ActiveMQConnectionFactory connectionFactory =
			new ActiveMQConnectionFactory (user, password, url);
		connection = connectionFactory.createConnection();
		connection.setExceptionListener (new ExceptionListener() {
			public void onException (JMSException exc) {
				exc.printStackTrace (System.err);
				System.exit (1);
			}
		});
		connection.start();
		
		// Create session
		session = connection.createSession(false,
				Session.AUTO_ACKNOWLEDGE);
		
		// Create a destination for subscribing the matrices 
		Destination destinationSubscribe = session.createTopic (inputTopic);
		
		//Create a destination for sending the result back to queue
		Destination destinationSender = session.createTopic (resultTopic);
		
		// Create a producer for sending the result
		producer = session.createProducer (destinationSender);
		
		// Create message consumer
		consumer = session.createConsumer (destinationSubscribe);
		
		System.out.println("Slave " + slaveId + " started.....");
		
		// Receive the input matrices as a ObjectMessage
		consumer.setMessageListener (new MessageListener() {
			public void onMessage (Message msg) {
				if (msg instanceof ObjectMessage) {
					try {
						MatrixInput matrix = (MatrixInput)
							((ObjectMessage)msg).getObject();
						
						// Get the complete 1st matrix 
						int input1[][] = matrix.getInput1();
						int noOfSlaves = matrix.getNoOfSlaves();
						
						// If the slave's id is within the range of
						// number of slaves requested to generate the
						// result then perform the operation.
						if(slaveId < noOfSlaves) {
							for(int colId = slaveId;
								colId < input1[0].length;
								colId += noOfSlaves) {
								
								// Get the the column matrix 
								int colMatrix[] =
									matrix.getColumnMatrix(colId);
								
								// Create a object of Compute Result
								ComputeResult result =
									new ComputeResult();
								
								// Get the resultant column object
								ResultantColumn rc = 
									result.computeResult(matrix, colMatrix,
														colId);
								
								// Publish the resultant object in result 
								// topic
								ObjectMessage objMsg = 
									session.createObjectMessage();
								objMsg.setObject((Serializable) rc);
								producer.send (objMsg);
							}
						}	
					}
					catch (JMSException exc) {
						System.out.printf ("Could not get message%n");
					}
				}
				else {
					System.out.printf ("Unknown message type%n");
				}
			}
		});
	}
	
	/**
	 * Parse an integer command line argument.
	 *
	 * @param  arg  Command line argument.
	 * @param  name  Argument name.
	 *
	 * @return  Integer value of <TT>arg</TT>.
	 *
	 * @exception  NumberFormatException
	 *     Thrown if <TT>arg</TT> cannot be parsed 
	 *     as an integer.
	 */
	private static int parseInt(String arg, String name) {
		try {
			return Integer.parseInt (arg);
		}
		catch (NumberFormatException exc) {
			System.err.println("MasterDriver: Invalid " + name + 
				": \"" + arg + "\"");
			usage();
			return 0;
		}
	}
	
	/**
	* Print a usage message and exit.
	 */
	private static void usage() {
		System.err.println("Slave Usage: java Slave <slaveid> <host> <port>");
		System.err.println("<slaveid> is the ID of the slave.");
		System.err.println("<host> is the Host name of the broker service.");
		System.err.println("<port> is the Port number of the broker service.");
		System.exit(1);
	}
}